package drds.binlog.client.adapter.support;

import drds.binlog.protocol.Message;
import drds.binlog.protocol.entry_protocol_proto.*;

import java.util.*;

/**
 * Message对象解析工具类
 */
public class MessageUtil
{

    public static void parse4Dml(Message message, Consumer<Dml> consumer)
    {
        if (message == null)
        {
            return;
        }
        List<Entry> entryList = message.getEntryList();

        for (Entry entry : entryList)
        {
            if (entry.getEntryType() == EntryType.TRANSACTIONBEGIN
                    || entry.getEntryType() == EntryType.TRANSACTIONEND)
            {
                continue;
            }

            RowChange rowChange;
            try
            {
                rowChange = RowChange.parseFrom(entry.getStoreValue());
            } catch (Exception e)
            {
                throw new RuntimeException("ERROR ## parser of eromanga-event has an error , data:" + entry.toString(),
                        e);
            }

            EventType eventType = rowChange.getEventType();

            final Dml dml = new Dml();
            dml.setDatabase(entry.getHeader().getSchemaName());
            dml.setTable(entry.getHeader().getTableName());
            dml.setType(eventType.toString());
            dml.setTs(System.currentTimeMillis());
            dml.setSql(rowChange.getSql());
            List<Map<String, Object>> data = new ArrayList<>();
            List<Map<String, Object>> old = new ArrayList<>();

            if (!rowChange.getIsDdl())
            {
                Set<String> updateSet = new HashSet<>();
                for (RowData rowData : rowChange.getRowDatasList())
                {
                    if (eventType != EventType.INSERT && eventType != EventType.UPDATE
                            && eventType != EventType.DELETE)
                    {
                        continue;
                    }

                    Map<String, Object> row = new LinkedHashMap<>();
                    List<Column> columns;

                    if (eventType == EventType.DELETE)
                    {
                        columns = rowData.getBeforeColumnsList();
                    } else
                    {
                        columns = rowData.getAfterColumnsList();
                    }

                    for (Column column : columns)
                    {
                        row.put(column.getName(),
                                JdbcTypeUtil.typeConvert(dml.getTable(),
                                        column.getName(),
                                        column.getValue(),
                                        column.getSqlType(),
                                        column.getMysqlType()));
                        // 获取update为true的字段
                        if (column.getUpdated())
                        {
                            updateSet.add(column.getName());
                        }
                    }
                    if (!row.isEmpty())
                    {
                        data.add(row);
                    }

                    if (eventType == EventType.UPDATE)
                    {
                        Map<String, Object> rowOld = new LinkedHashMap<>();
                        for (Column column : rowData.getBeforeColumnsList())
                        {
                            if (updateSet.contains(column.getName()))
                            {
                                rowOld.put(column.getName(),
                                        JdbcTypeUtil.typeConvert(dml.getTable(),
                                                column.getName(),
                                                column.getValue(),
                                                column.getSqlType(),
                                                column.getMysqlType()));
                            }
                        }
                        // update操作将记录修改前的值
                        if (!rowOld.isEmpty())
                        {
                            old.add(rowOld);
                        }
                    }
                }
                if (!data.isEmpty())
                {
                    dml.setData(data);
                }
                if (!old.isEmpty())
                {
                    dml.setOld(old);
                }
                consumer.accept(dml);
            }
        }
    }

    public interface Consumer<T>
    {

        void accept(T t);
    }
}
